package cn.ymatrix.mqttToYMatrix;

import cn.ymatrix.utils.Config;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.HttpClientBuilder;
import org.eclipse.paho.client.mqttv3.*;
import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

public class MqttDataReceiverTOYMatrix {
  public static void main(String[] args) {
    String broker = Config.getMqttBroker();
    String clientId = Config.getMqttClientId();
    String topic = Config.getMqttTopic();

    try {
      MqttClient client = new MqttClient(broker, clientId);
      MqttConnectOptions connOpts = new MqttConnectOptions();
      client.connect(connOpts);

      // 设置回调以处理接收到的MQTT消息
      client.setCallback(
          new MqttCallback() {
            public void connectionLost(Throwable cause) {
              System.out.println("Connection lost: " + cause.getMessage());
            }

            public void messageArrived(String topic, MqttMessage message) throws Exception {
              // 处理接收到的MQTT消息
              String receivedData = new String(message.getPayload());
              // 将数据写入到YMatrix(JDBC和HTTP方式)
              //  jdbcWriteToYMatrix(receivedData);
              sendHttpPostRequest(receivedData);
            }

            // 成功返回的方法
            public void deliveryComplete(IMqttDeliveryToken token) {
              try {
                System.out.println("send success data:" + token.getMessage().getPayload());
              } catch (MqttException e) {
                throw new RuntimeException(e);
              }
            }
          });

      // 订阅MQTT主题
      client.subscribe(topic);

      // 等待接收消息
      while (true) {
        Thread.sleep(100);
      }
    } catch (MqttException | InterruptedException e) {
      e.printStackTrace();
    }
  }

  // JDBC 写入
  private static void jdbcWriteToYMatrix(String receivedData) throws SQLException {
    String copySql =
        "COPY "
            + Config.getTableName()
            + " FROM STDIN DELIMITER '"
            + Config.getDelimiter()
            + "' CSV";
    Connection connection = null;
    InputStream inputStream = null;
    try {
      connection =
          DriverManager.getConnection(Config.getUrl(), Config.getUserName(), Config.getPassWord());
      CopyManager copyManager = new CopyManager((BaseConnection) connection);
      inputStream = new ByteArrayInputStream(receivedData.getBytes(StandardCharsets.UTF_8));
      copyManager.copyIn(copySql, inputStream);
      inputStream.close();
      connection.close();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  // http 写入
  private static void sendHttpPostRequest(String receivedData) {
    HttpClient httpClient =
        HttpClientBuilder.create().setMaxConnPerRoute(Config.getMaxConnPerRoute()).build();
    HttpPost httpPost = new HttpPost(Config.getHttpUrl());
    String dataStringBuilder = Config.getTableName() + "\n" + receivedData;
    try {
      httpPost.setEntity(new StringEntity(dataStringBuilder, StandardCharsets.UTF_8));
      httpPost.setHeader("Content-Type", "text/plain");
      httpClient.execute(httpPost);
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
}
